#pragma once
#include "../common/net.hpp"
#include "../common/message.hpp"
#include <unordered_set>
#include <unordered_map>

//主题创建->主题不存在则创建,存在则返回OK
//主题的删除
//主题的订阅->服务端将订阅者管理起来,客户端将订阅主题管理起来
//主题的取消订阅
//主题消息的发布


//发布订阅:多个客户端与一个服务端实现中间数据转发:多个消息发布客户端,中转服务器,多个消息处理客户端
//对外提供一个主题操作消息处理回调函数->Dispatcher模块->主题消息,分发给回调函数
//数据的管理:订阅者信息管理 hash_map<connection,订阅者> 某个订阅者断开连接,删除关联关系
//hash_map<主题,订阅者>:某个主题收到消息,将消息发送给订阅者
//订阅者包含通信连接,所订阅的主题名称
//针对连接断开的处理回调函数

namespace rpc
{
    namespace server
    {
        //主题管理
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            TopicManager(){}

            //收到主题消息进行操作
            void onTopicRequest(const BaseConnection::ptr & conn,const TopicRequest::ptr & msg)
            {
                //主题创建/主题删除/主题订阅/主题取消订阅/主题消息的发布
                TopicOptype topic_optype = msg->optype();
                bool ret = true;
                switch(topic_optype)
                {
                    case TopicOptype::TOPIC_CREATE: {topicCreate(conn,msg);break;}//主题创建
                    case TopicOptype::TOPIC_REMOVE: {topicRemove(conn,msg);break;}//主题删除
                    case TopicOptype::TOPIC_SUBSCRIBE: {ret = topicSubscribe(conn,msg);break;}//主题订阅
                    case TopicOptype::TOPIC_CANCEL: {topicCancel(conn,msg);break;}//取消订阅
                    case TopicOptype::TOPIC_PUBLISH: {ret = topicPublish(conn,msg);break;}//消息发布
                    default: return errorResponse(conn,msg,RCode::RCODE_INVALID_OPTYPE);
                }
                if(!ret) return errorResponse(conn,msg,RCode::RCODE_NOT_FOUND_TOPIC);
                return topicResponse(conn,msg);
            }

            //链接断开进行处理->一个订阅者在连接断开的时候,删除其管理的数据
            void onShutdown(const BaseConnection::ptr & conn)
            {
                //消息发布者断开连接,不需要操作,消息订阅者断开连接,需要删除管理数据
                std::vector<Topic::ptr> topics;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _subscribers.find(conn);
                    if(it == _subscribers.end())//断开连接的不是订阅者
                    {
                        return;
                    }
                    subscriber = it->second;
                    for(auto & topic_name:subscriber->topics)
                    {
                        auto topic_it = _topics.find(topic_name);
                        if(topic_it == _topics.end()) continue;
                        topics.push_back(topic_it->second);//获取到订阅者订阅的主题
                    }
                    _subscribers.erase(it);//删除订阅者
                }
                //主题中移除订阅者
                for(auto & topic:topics)
                {
                    topic->removeSubscriber(subscriber);
                }
            }

        private:
            //出错响应
            void errorResponse(const BaseConnection::ptr & conn,const TopicRequest::ptr & msg,RCode rcode)
            {
                auto msg_rsp=MessageFactory::create<TopicResponse>();
                msg_rsp->setId(msg->rid());
                msg_rsp->setMType(MType::RSP_TOPIC);
                msg_rsp->setRCode(rcode);
                conn->send(msg_rsp);
            }

            //正确响应
            void topicResponse(const BaseConnection::ptr & conn,const TopicRequest::ptr & msg)
            {
                auto msg_rsp = MessageFactory::create<TopicResponse>();
                msg_rsp->setId(msg->rid());
                msg_rsp->setMType(MType::RSP_TOPIC);
                msg_rsp->setRCode(RCode::RCODE_OK);
                conn->send(msg_rsp);
            }
            
            //主题的创建
            void topicCreate(const BaseConnection::ptr &conn,const TopicRequest::ptr & msg)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                std::string topic_name = msg->topicKey();
                auto topic = std::make_shared<Topic>(topic_name);
                _topics.insert(std::make_pair(topic_name,topic));
            }

            //主题的删除
            void topicRemove(const BaseConnection::ptr & conn,const TopicRequest::ptr & msg)
            {
                //删除订阅者,从订阅者中将主题删除掉
                //删除主题->主题名称和主题对象的映射关系
                std::string topic_name = msg->topicKey();
                //找到会被影响的订阅者
                std::unordered_set<Subscriber::ptr> subscribers;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _topics.find(topic_name);
                    if(it == _topics.end())
                    {
                        return;
                    }
                    subscribers = it->second->subscribers;
                    _topics.erase(it);
                }
                for(auto & subscriber:subscribers)
                {
                    subscriber->removeTopic(topic_name);
                }
            }

            //主题订阅
            bool topicSubscribe(const BaseConnection::ptr & conn,const TopicRequest::ptr & msg)
            {
                //找出主题对象和订阅者对象
                //在主题对象中,新增一个订阅者关联的链接 在订阅者对象中,新增一个订阅的主题
                //没有找到主题->出错返回,没有找到订阅者对象,新增一个订阅者
                Topic::ptr topic;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(msg->topicKey());
                    if(topic_it == _topics.end())
                    {
                        return false;
                    }
                    topic= topic_it ->second;
                    auto sub_it = _subscribers.find(conn);
                    if(sub_it != _subscribers.end())
                    {
                        subscriber = sub_it->second;
                    }
                    else
                    {
                        subscriber = std::make_shared<Subscriber>(conn);
                        _subscribers.insert(std::make_pair(conn,subscriber));
                    }
                }
                topic->appendSubscriber(subscriber);
                subscriber->appendTopic(msg->topicKey());
                return true;
            }

            //取消订阅
            void topicCancel(const BaseConnection::ptr & conn,const TopicRequest::ptr & msg)
            {
                //找出主题对象和订阅者对象
                Topic::ptr topic;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(msg->topicKey());
                    if(topic_it != _topics.end())
                    {
                        topic = topic_it->second;
                    }
                    auto sub_it = _subscribers.find(conn);
                    if(sub_it != _subscribers.end())
                    {
                        subscriber = sub_it->second;
                    }
                }
                if(subscriber) subscriber->removeTopic(msg->topicKey());
                if(topic && subscriber) topic->removeSubscriber(subscriber);
            }

            //主题的消息发送
            bool topicPublish(const BaseConnection::ptr & conn,const TopicRequest::ptr & msg)
            {
                Topic::ptr topic;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(msg->topicKey());
                    if(topic_it == _topics.end())
                    {
                        return false;
                    }
                    topic=topic_it ->second;
                }
                topic->pushMessage(msg);
                return true;
            }
         private:
            //订阅者信息
            struct Subscriber
            {
                using ptr = std::shared_ptr<Subscriber>;
                std::mutex _mutex;
                BaseConnection::ptr conn;//通信链接
                std::unordered_set<std::string> topics;//一个订阅者订阅的多个主题
                Subscriber(const BaseConnection::ptr & c):conn(c){}

                //添加订阅主题
                void appendTopic(const std::string & topic_name)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    topics.insert(topic_name);
                }

                //主题被删除或取消订阅的时候
                void removeTopic(const std::string & topic_name)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    topics.erase(topic_name);
                }
            };

            //主题相关信息
            struct Topic
            {
                using ptr = std::shared_ptr<Topic>;
                std::mutex _mutex;
                std::string topic_name;//主题名称
                std::unordered_set<Subscriber::ptr> subscribers;//主题有哪些人已经订阅
                Topic(const std::string & name):topic_name(name){}

                //新增订阅->添加订阅者
                void appendSubscriber(const Subscriber::ptr & subscriber)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    subscribers.insert(subscriber);
                }

                //取消订阅或订阅者链接断开,删除一个订阅者
                void removeSubscriber(const Subscriber::ptr & subscriber)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    subscribers.erase(subscriber);
                }

                //收到消息发布,推送消息
                void pushMessage(const BaseMessage::ptr & msg)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    for(auto & subscriber:subscribers)//广播发送
                    {
                        subscriber->conn->send(msg);
                    }
                }
            };
        private:
            std::mutex _mutex;
            std::unordered_map<std::string,Topic::ptr> _topics;//主题名称:主题
            std::unordered_map<BaseConnection::ptr,Subscriber::ptr> _subscribers;//连接:订阅者
        };
    }
}